package MQ

import (
	"github.com/nats-io/nats.go"
	"log"
	"rpcSrv/configs"
	"rpcSrv/utils"
	"sync"
	"time"
)

var (
	conn *nats.Conn
	once sync.Once
)

func initMQ()  {
	once.Do(func() {
		url := configs.GetNatsUrlConf()
		var err error
		conn, err = nats.Connect(url)
		if utils.CheckErr(err) {
			log.Fatal("连接nats出错！")
		}
	})
}

type QueueOptions struct {
	Async     bool          //true就是并发处理  false同步处理
	TimeLimit time.Duration //设置等待时间限制并发速度
	RcvNum    int
}

type mq struct {
	subj     string
	receiver func(msg []byte) error //消费者处理函数
	Opt      QueueOptions           //主要用于并发限速
}

func NewMQ(subj string, rcv func(msg []byte) error, opt ...QueueOptions) *mq {
	initMQ()
	o := new(mq)
	o.subj = subj
	o.receiver = rcv
	if len(opt) > 0 {
		o.Opt = opt[0]
	}
	if o.Opt.RcvNum > 1 {
		for i := 0; i < o.Opt.RcvNum; i++ {
			go o.listen()
		}
	} else {
		go o.listen()
	}

	return o
}

func (this *mq) rcv(msg []byte) {
	err := this.receiver(msg)
	if utils.CheckErr(err) {
		//保存数据库
	}
}

func (this *mq) listen() {
	sub, err := conn.QueueSubscribeSync(this.subj, "queue")
	if utils.CheckErr(err) {
		time.Sleep(time.Second)
		this.listen()
		return
	}
	for {
		msg, err := sub.NextMsg(time.Minute*3)
		if utils.CheckErr(err) {
			//保存数据库
			continue
		}
		if this.Opt.Async {
			go this.rcv(msg.Data)
		} else {
			this.rcv(msg.Data)
		}
		if this.Opt.TimeLimit > 0 {
			time.Sleep(this.Opt.TimeLimit)
		}
	}
}

func (this *mq) Publish(msg []byte) error {
	return conn.Publish(this.subj,msg)
}





